package net.xuele.learn.flink.cdc;

//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.connector.jdbc.JdbcDataTable;
//import org.apache.flink.connector.jdbc.utils.JDBCUtils;
//import org.apache.flink.configuration.ConfigConstants;
//import org.apache.flink.configuration.Configurations;
//import org.apache.flink.data.connector.elasticsearch.ElasticsearchUtils;
//import org.apache.flink.data.connector.jdbc.JdbcFormat;
//import org.apache.flink.data.connector.jdbc.JdbcSource;
//import org.apache.flink.data.connector.jdbc.JdbcTarget;
//import org.apache.flink.data.connector.jdbc.TupleDataModel;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.connectors.jdbc.JdbcBatchEnvironment;
//import org.apache.flink.streaming.connectors.jdbc.JdbcEnvironment;
//import org.apache.flink.table.api.TableSchema;
//import org.apache.flink.table.api.dataview.MapView;
//import org.apache.flink.table.data.ArrayData;
//import org.apache.flink.table.data.JdbcDataTable;
//import org.apache.flink.table.data.RowData;
//import org.apache.flink.table.data.TimestampKind;
//import org.apache.flink.table.sources.StreamTableSource;
//import org.apache.flink.table.sources.jdbc.JdbcTableSource;
//import org.apache.flink.table.sources.jdbc.TupleStreamSource;
//import org.apache.flink.table.sources.jdbc.TypedTableSource;
//import org.apache.flink.table.sources.jdbc.FixedTimeWindowSource;
//import org.apache.flink.table.sources.jdbc.PartitionKeyColumn;
//import org.apache.flink.table.utils.JdbcUtils;
//import org.apache.flink.types.Row;
//import org.apache.flink.types.TupleType;

import java.util.Arrays;

public class FlinkEcsToClickHouseExample {

    public static void main(String[ ] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        // Configure Elasticsearch Connector
//        ElasticsearchUtils esUtils = new ElasticsearchUtils(env);
//        JdbcFormat<String> jdbcFormat = JdbcFormat.VARCHAR;
//        configs.setConfig(ConfigConstants.CONFIG, config);
//
//        // Configure ClickHouse Connector
//        configs.setConfig(ConfigConstants.CONFIG, config);
//
//        // Set up Jdbc Table Source
//        TableSchema tableSchema = new TableSchema("ClickHouse", jdbcFormat);
//        TableSchema jdbcSchema = new TableSchema("jdbc", jdbcFormat);
//        TupleDataModel tupleDataModel = new TupleDataModel();
//        ArrayData arrayData = new ArrayData(Arrays.asList("1", "2", "3"));
//        TupleStreamSource<RowType> tupSource = new TypedTableSource<>(tableSchema, tupleDataModel, arrayData, JdbcUtils.parse(jdbcSchema.getJDBC()));
//
//        // Set up ClickHouse Source
//        TupleDataModel cqlTableDataModel = new TupleDataModel();
//        FixedTimeWindowSource<RowType> cqlSource = new FixedTimeWindowSource<>(tableSchema, cqlTableDataModel, RowType.ROW, "0 sec");
//
//        // Set up Jdbc Table Sink
//        JdbcBatchEnvironment batchEnvironment = new JdbcBatchEnvironment(config);
//        batchEnvironment.addTableSource(tupSource);
//        batchEnvironment.addTableSource(cqlSource);
//
//        // Run ClickHouse Batch Environment
//        batchEnvironment.start();
//
//        // Collect results
//        DataStream<RowType> clickHouseStream = batchEnvironment.toStream();
//        DataStream<RowType> esResultStream = esUtils.executeEsToClickHouse(clickHouseStream);
//
//        // Join results
//        MapView<RowType, JdbcDataTable> results = esResultStream.map(
//                new JdbcDataTable -> {
//
//            JdbcDataTable esJdbcTable = (JdbcDataTable) JdbcUtils.convertToTable(esUtils.getJdbcDataTable(new JdbcDataTable[ ] {}));
//
//            JdbcDataTable clickHouseTable = JdbcUtils.convertToTable(clickHouseStream.getData());
//            RowData esRowData = JdbcUtils.convertToRowData(esJdbcTable.getSchema(), esJdbcTable.getData());
//            RowData clickHouseRowData = JdbcUtils.convertToRowData(clickHouseTable.getSchema(), clickHouseStream.getData());
//            JdbcDataTable mergedJdbcTable = new JdbcDataTable(clickHouseRowData.size());
//            for (int i = 0; i < esRowData.size(); i++) {
//                mergedJdbcTable.addRow(new RowType(clickHouseRowData.getFieldNames()[i], clickHouseRowData.get(i)));
//            }
//            return mergedJdbcTable;
//        });
//
//        // Save results to ClickHouse database
//        JdbcBatchEnvironment clickHouseBatchEnvironment = new JdbcBatchEnvironment(config);
//        clickHouseBatchEnvironment.addTableSource(results);
//        clickHouseBatchEnvironment.start();
//    }
//
//    private static JdbcConfig config;
//
//    public FlinkEcsToClickHouseExample(JdbcConfig config) {
//        this.config = config;
//    }
    }
}